package dm

import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}

object dm_bi_audition_conversion {

  def main(args: Array[String]): Unit = {

      //创建上下文环境配置对象
      val conf: SparkConf = new SparkConf().setAppName("dm.dm_bi_audition_conversion")
      //创建SparkSession对象
      val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()
      //RDD=>DataFrame=>DataSet转换需要引入隐式转换规则，否则无法转换
      //spark不是包名，是上下文环境对象名
      import spark.implicits._
      // TODO   测试 dm_bi_audition_conversion 存过使用 sparksql运行
      var sql =
        """
        insert overwrite table tmp.dm_bi_audition_conversion partition(dt)
        select
                `时间维度`,
                  case when t.holiday_flag = '1' then '是' else '否' end `是否休息日`,
                `学院`,
                `分类`,
                `课程名称`,
                `章节名称`,
                `课程时间` ,
                `讲师姓名`,
                `试听课人数`,
                `报名人数`,
                  -- count(distinct dd.user_id) / count(distinct st.dapeng_user_id) `试学课转化率`,
                `试学课销售金额`,
                st.op_time,
                `标识`,
                st.dt
            from(
              select
                  substr(st.audition_date,1,10) `时间维度`,
                  st.college_name `学院`,
                  `分类`,
                  li.course_title `课程名称`,
                  li.title `章节名称`,
                  `课程时间` ,
                  stff.name `讲师姓名`,
                  count(distinct st.dapeng_user_id) `试听课人数`,
                  count(distinct case when substr(dd.first_pay_time,1,10) <= date_add(last_day(st.audition_date),3) then dd.user_id else null end) `报名人数`,
                  -- count(distinct dd.user_id) / count(distinct st.dapeng_user_id) `试学课转化率`,
                  sum(distinct case when substr(dd.first_pay_time,1,10) <= date_add(last_day(st.audition_date),3) then dd.all_pay else 0 end) `试学课销售金额`,
                  from_unixtime(unix_timestamp(),'yyyy-MM-dd HH:mm:ss') op_time,
                  '$exe_id' `标识`,
                  substr(st.audition_date,1,7) dt
              from(
                  select
                      case when `分类` = '微信2' then '小鹏美术' else s.college_name end college_name,`分类`,s.college_id,s.dapeng_user_id,s.curriculum_id,s.audition_month,s.audition_date,s.first_listen_time,s.lession_start_time,
                      case when s.first_listen_time = ma then 1 else 0 end flag,`课程时间`
                  from(
                      select
                      distinct
                      x.college_name,`分类`,st.college_id,st.dapeng_user_id,st.curriculum_id,substr(st.audition_date,1,7) audition_month,st.audition_date,st.first_listen_time,st.lession_start_time,
                      max(first_listen_time) over(partition by st.dapeng_user_id,x.college_name) ma,`课程时间`
                      from(
                              select
                              college_id,dapeng_user_id,curriculum_id,substr(audition_date,1,7) audition_month,audition_date,coalesce(first_listen_time,lession_start_time) first_listen_time,lession_start_time,
                              case
                                when hour(lession_start_time) < 13 then '10:00'
                                when hour(lession_start_time) >= 13 and hour(lession_start_time) < 18 then '14:30'
                                else '19:30'
                              end  `课程时间`,
                              CASE
                                  WHEN college_id='j5m484vz'
                                  AND course_id ='iipkwky9' THEN 'QQ'
                                  WHEN college_id='j5m48deg'
                                  AND course_id ='kdijfswalv' THEN '微信2'
                                  else '微信'
                              END `分类`
                              from dws.dws_attend_lecturesn where course_type = 'TRIAL'  -- and  lecturesn_type = 'live' and student_name NOT like 'vip%' and student_name NOT like 'VIP%'
                      ) as st
                      left join (select order_category,first_pay_date,user_id,all_pay,first_pay_time from dm.dm_order where flag = 'Y' and user_id <>'' and user_id is not null ) as dd
                          on st.dapeng_user_id = dd.user_id and st.college_id = dd.order_category -- and st.audition_month = substr(dd.first_pay_date,1,7)
                      left join dim.dim_college as x
                          on x.college_id = st.college_id
                      where st.first_listen_time < dd.first_pay_time or dd.first_pay_date is null
                  ) as s
              ) as st
              left join dim.dim_live_curriculum as li
                  on li.curriculum_id = st.curriculum_id and li.flag = 'Y'
              left join (select pc_user_id,name from dim.v_dim_staff_info where pc_user_id <> '' and pc_user_id is not null)as stff
                  on li.teacher_id = stff.pc_user_id
              left join (select order_category,first_pay_date,user_id,all_pay,first_pay_time from dm.dm_order where flag = 'Y' and user_id <>'' and user_id is not null ) as dd
                  on st.dapeng_user_id = dd.user_id and st.college_id = dd.order_category and st.flag = 1 --and st.audition_month = substr(dd.first_pay_date,1,7)
              group by
                  substr(st.audition_date,1,10),
                  st.college_name,
                  `分类`,
                  li.course_title ,
                  li.title ,
                  `课程时间`,
                  stff.name,
                  substr(st.audition_date,1,7)
              ) as st
            left join dim.dim_date as t
               on t.date_type = st.`时间维度`
          """
     spark.sql("set hive.exec.dynamici.partition=true")
     spark.sql("set hive.exec.dynamic.partition.mode=nonstrict")
     spark.sql("set spark.sql.adaptive.enabled=true")
     val frame: DataFrame = spark.sql(sql)
     frame.show()
      //释放资源
      spark.stop()
  }
}

artifactId